package com.atguigu.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.util.DimUtil;
import com.atguigu.util.HBaseDataSourceUtil;
import com.atguigu.util.JedisUtil;
import com.atguigu.util.ThreadPoolUtil;
import lombok.SneakyThrows;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

public abstract class DimInfoAsyncFunction_Thread<T> extends RichAsyncFunction<T, T> {

    private HBaseDataSourceUtil hBaseDataSourceUtil;
    private ThreadPoolExecutor threadPoolExecutor;
    private JedisPool jedisPool;
    private String tableName;

    public DimInfoAsyncFunction_Thread(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        hBaseDataSourceUtil = new HBaseDataSourceUtil();
        threadPoolExecutor = ThreadPoolUtil.getThreadPoolExecutor();
        jedisPool = JedisUtil.getJedisPool();
    }

    public abstract String getPk(T input);

    public abstract void join(T input, JSONObject dimInfo);

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
        threadPoolExecutor.execute(new Runnable() {
            @SneakyThrows
            @Override
            public void run() {
                Connection connection = hBaseDataSourceUtil.getConnection();
                Jedis jedis = jedisPool.getResource();
                JSONObject dimInfo = DimUtil.getDimInfo(connection, jedis, tableName, getPk(input));
                join(input, dimInfo);
                jedis.close();
                hBaseDataSourceUtil.addBack(connection);
                resultFuture.complete(Collections.singletonList(input));
            }
        });
    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        System.out.println("TimeOut>>>>>>" + input);
    }
}
